package com.atguigu.champter5.Flink05.sink;

import com.atguigu.beans.WaterSensor;
import com.atguigu.source.RandomWaterSenson;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisConfigBase;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

public class Flink03_Sink_Redis {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        FlinkJedisConfigBase flinkJedisConfigBase = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop102")
                .setPort(6379)
                .setDatabase(0)
                .build();

        env
                .addSource(new RandomWaterSenson())
                .addSink(new RedisSink<WaterSensor>(flinkJedisConfigBase, new RedisMapper<WaterSensor>() {

                    @Override
                    public RedisCommandDescription getCommandDescription() {
                        return new RedisCommandDescription(RedisCommand.HSET, "sensor");
                    }

                    @Override
                    public String getKeyFromData(WaterSensor data) {
                        return data.getId();
                    }

                    @Override
                    public String getValueFromData(WaterSensor data) {
                        return data.toString();
                    }
                }));

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
